package study

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * @Author: 蔡翔
  * @Date: 2019/9/3 20:00
  * @Version 1.0
  *
  *    滑动窗口：每隔1s，统计最近2s内的数据，打印到控制台
  */
object SocketWindowWordCount {
  def main(args: Array[String]): Unit = {
    //获取端口号
    val port: Int =try {
      import org.apache.flink.api.java.utils.ParameterTool
      ParameterTool.fromArgs(args).getInt("port")   //如果解析成功 就直接把ParameterTool.fromArgs(args).getInt("port") 赋值给port
    }catch{
      case e:Exception=>{
        System.err.println("No Port set.use default 9000")
      }
        9000   //解析失败后直接把 9000赋值给port
    }


    //1.获取运行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //2.连接socket 获取输入数据
    val text = env.socketTextStream("192.168.2.125",port,'\n')

    /*//隐式转换
    import org.apache.flink.api.scala._

    //解析数据(把数据打平)，分组，窗口计算，并且聚合求sum ;;line=>line.split("\\s") 把line切开 并且返回,返回给map，在让map做处理，处理好后又return
    val windowCount = text.flatMap(line => line.split("\\s"))  //打平，把每一行单词都切开
      .map(w=>WordWithCount(w,1))  //把单词转成word,1 这种形式
      .keyBy("word")  //分组
      .timeWindow(Time.seconds(2),Time.seconds(1)) //指定窗口大小，指定间隔时间
      .sum("count")
      //.reduce((a,b)=>WordWithCount(a.word,b.count+b.count))
    windowCount.print().setParallelism(1)
    env.execute("Socket window job")*/
  }

  case class WordWithCount(word:String ,count: Long)  //这就是一个类，默认生成get 和set方法。
}

